import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Hashtable;
import java.util.Map;
import java.util.TreeMap;
import java.io.*;
import java.net.*;
public class Cloud implements Runnable{

	private String name;
	private int portID;
	private ServerSocket broker;
	private int numDataCenters;
	private ArrayList<Server> serverList;
	private ArrayList<DataCenter> dcList;
	private ArrayList<VirtualMachine> vmTypeList;
	private Hashtable<Integer,Integer []> activeVmList;
	private int vmCtr=0;
	
	public Cloud(String name, ArrayList<DataCenter> dcList, ArrayList<VirtualMachine> vmTypeList, int portID){
		this.name = name;
		serverList = new ArrayList<Server>();
		activeVmList = new Hashtable<Integer, Integer[]>();
		this.numDataCenters=0;
		this.vmTypeList = vmTypeList;
		this.dcList=dcList;
		this.portID = portID;	
		try {			
			this.start();

		} catch (IOException e) {
			throw new RuntimeException("Cannot open port ", e);
		} catch (InterruptedException e) {
            e.printStackTrace();
		}			

	}
		
	public void start() throws IOException, InterruptedException {
		runWorkflowJob("src\\workflow\\workflow1.csv");
		broker = new ServerSocket(this.portID);
		launchAllServers();
		Thread.sleep(200);
		printAvailableResources();

	}
	
	public void stop() throws IOException {

		terminateAllServers();
		System.out.println("Shutting down Cloud "+this.name);
		this.broker.close();
	}

    public void run() {
		
		boolean running=true;
		String input;
		LoadBalancer lb = new LoadBalancer(vmTypeList);
		
		try {
			while (running) {
	    		Socket connectionSocket = broker.accept();
	    		ObjectInputStream inFromSystem =
	       			new ObjectInputStream(connectionSocket.getInputStream());
		        ObjectOutputStream outToSystem = new ObjectOutputStream(connectionSocket.getOutputStream());
		        boolean connected=true;
		        
		        while (connected) {
		    		input = (String) inFromSystem.readObject();
		    		System.out.println("Input from administrator:"+input);	
		    		if (input.equals("Shutdown")) { 
			            outToSystem.writeObject(new String("Shutting down "+this.name+'\n'));
			            running=false;
			            connected=false;
		    		} else if(input.equals("Goodbye")) {
			            outToSystem.writeObject(new String("Goodbye from "+this.name+'\n'));
			            connected=false;
		    		} else if(input.equals("Hello")) {
		    		    outToSystem.writeObject(new String("Welcome to "+this.name+"\n"));
		    		} else if(input.equals("ListVms")) {
		    			String output=listVMs();
		    		    outToSystem.writeObject(output);
		    		} else if(input.equals("RequestVms")) {			    			
	    				int [] vmRequest = (int []) inFromSystem.readObject();
	    				ArrayList<Integer> vmidList=new ArrayList<Integer>();
	    				double [] cost={0};
	    				String output=requestVMs(vmRequest,lb,vmidList,cost);
		    		    outToSystem.writeObject(output);
		    		    
		    		    //get user response as to purchase or cancel these vms
		    		    System.out.println("vmidList size="+vmidList.size());
		    		    String userResponse=(String) inFromSystem.readObject();	
		    		    if(userResponse.equals("PurchaseVms")) {
			    			output=purchaseVMs(vmidList,cost);
			    		    outToSystem.writeObject(output);
			    		    printAvailableResources();					    							    				
		    		    
		    		    } else if(userResponse.equals("CancelVms")) {
			    			output=cancelVMs(vmidList);		    		    	
			    		    outToSystem.writeObject(output);
			    		    printAvailableResources();	
		    		    }		    			
		    		} else if(input.equals("ConnectVm")) {	
	    				Integer vmid = (Integer) inFromSystem.readObject();
	    				String output=connectToVM(vmid);	    					
	    				outToSystem.writeObject(output);
	    				connected=false;
		    		} else if (input.equals("DisconnectVm")) {
	    				Integer vmid = (Integer) inFromSystem.readObject();
	    				String output=disconnectFromVM(vmid);	    					
	    				outToSystem.writeObject(output);
	    				connected=false;
		    		} else if (input.equals("SubmitJob")) {
	    				Job jobRequest = (Job) inFromSystem.readObject();
	    				ArrayList<Integer> vmidList=new ArrayList<Integer>();
	    				double [] cost={0};
	    				String output= submitJob(jobRequest,lb,vmidList,cost);
		    		    outToSystem.writeObject(output);
		    		    
		    		    //get user response as to purchase or release these vms
		    		    String userResponse=(String) inFromSystem.readObject();	
		    		    if(userResponse.equals("PurchaseVms")) {
			    			output=purchaseVMs(jobRequest,cost);		    		    	
			    		    outToSystem.writeObject(output);
			    		    outToSystem.writeObject(jobRequest);			    		    
			    		    printAvailableResources();					    							    						    		    
		    		    } else if(userResponse.equals("CancelVms")) {
			    			output=cancelVMs(vmidList);		    		    	
			    		    outToSystem.writeObject(output);
			    		    printAvailableResources();	
		    		    }		    		    		    		    	    				
	    				connected=false;
		    		} else if (input.equals("ReleaseVm")) {
	    				Integer vmid = (Integer) inFromSystem.readObject();
		    			String output= releaseVM(vmid);	    					
	    				outToSystem.writeObject(output);
		    		    printAvailableResources();
	    			}
		        }
	        	inFromSystem.close();
	            outToSystem.close();		    		
			}
			this.stop();
		} catch (IOException e) {
            throw new RuntimeException(
            		"Error accepting connection", e);				
		} catch (ClassNotFoundException e) {
            throw new RuntimeException(
            		"Object class incompatable", e);				
		}
    }
	
	public int launchAllServers() {
		int serverId=this.portID+1;
		
		for (int i=0; i<dcList.size(); i++) {
			DataCenter dc=dcList.get(i);
			numDataCenters++;			
			ArrayList<PhysicalMachine> pmList=dc.getPmList();
			for (int j=0; j<pmList.size(); j++) {
				PhysicalMachine pm=pmList.get(j);
				this.launchServer(dc,pm,serverId);
				serverId++;
			}
		}
		return 0;
	}
	
	public int terminateAllServers() {
		while (serverList.size() > 0) {
			terminateServer(serverList.get(0).getid());
		}
		return 0;
	}
	
	public int launchServer(DataCenter dc, PhysicalMachine pm, int serverId) {
		
		Server newServer = new Server(dc, pm, vmTypeList, serverId);
		serverList.add(newServer);

		return 0;
	}

	public int terminateServer(int serverID) {
		int error = -1;	// Default to server not found
		
		for (int i=0; i<serverList.size(); i++){
			if (serverList.get(i).getid() == serverID) {
				serverList.get(i).stop();
				serverList.remove(i);
				error = 0;
				break;
			}
		}
		return error;
	}

	public void printAvailableResources()
	{
		System.out.println(this.name+" available resources:");
		for (Server server: serverList) 
			server.printAvailableResources();
		
	}
	
	public void setServerList(ArrayList<Server> serverList) {
		this.serverList = serverList;
	}

	public ArrayList<Server> getServersList() {
		return serverList;
	}
  
	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public int getNumDataCenters() {
		return numDataCenters;
	}
	
	public int getNumServers() {
		return serverList.size();
	}
	
	/**
	 * method listVMs: lists vm types available on the cloud
	 * @return
	 */
	private String listVMs() {
		
		String output;
		output = "Available VM Types Are #" + 
   	     		"Name:\t CPU\t Memory(GB)\t Disk(GB)\t Price/Hour #";
		for (VirtualMachine vm : vmTypeList) {
			output += vm.getname() + ":\t " + vm.getvcpu() + "\t " + vm.getmemory()+
					"\t\t " + vm.getdisk() + "\t\t $ " + vm.getprice() + "#";
		}
		return output;
    }
	
	/**
	 * method requestVMs: processes client request for VMs
	 * @param vmRequest
	 * @param lb
	 * @param vmidList
	 * @param cost
	 * @return: output string
	 */
	private String requestVMs(int [] vmRequest, LoadBalancer lb, ArrayList<Integer> vmidList, double [] cost) {
		
		int id0=this.vmCtr;
		int nVms=0;
		for (int i=0; i<vmTypeList.size(); i++) {
			nVms+=vmRequest[i];
		}
		this.vmCtr+=nVms;
		
		//call load balancer for vm placement
		ArrayList<Integer []> serverAllocation=lb.vmPlacement(vmRequest, serverList,id0);
		System.out.println("Server Allocation:"+serverAllocation.size());
		
		//estimate cost of assigned vms
		int [] vmAvailable= new int[vmTypeList.size()];	
		while(serverAllocation.size()>0) {
			Integer [] a= serverAllocation.remove(0);
			cost[0]+=vmTypeList.get(a[1]).getprice();	
			vmAvailable[a[1]]++;
			Integer [] vmOpts= {a[0], a[1], 0};
			activeVmList.put(a[2], vmOpts);
			vmidList.add(a[2]);			    					
		}
		String output="";
		for (int i=0; i<vmTypeList.size(); i++)
			if(vmRequest[i]>0)
				output+=" VMtype :"+vmTypeList.get(i).getname()+" requested:" +vmRequest[i]+" available: "+vmAvailable[i] +"#";
		
		DecimalFormat df = new DecimalFormat("0.00");							
		output+="# Total cost for vms: $"+Double.valueOf(df.format(cost[0]))+"#";
		System.out.println("vmidList size "+vmidList.size());
		return output;
		
	}
	
	/**
	 * method purchaseVMs: processes client purchase of VMs for a vm request
	 * @param vmidList
	 * @param cost
	 * @return: output string
	 */
	private String purchaseVMs(ArrayList<Integer> vmidList, double [] cost) {

		String output="";
		while(vmidList.size()>0) {
			Integer vmid=vmidList.remove(0);
			Integer [] vmOpts=activeVmList.get(vmid);
			output+="VMtype :"+vmTypeList.get(vmOpts[1]).getname()+" launched with vmid: "+vmid+"#";
		}
		
		DecimalFormat df = new DecimalFormat("0.00");							
		output+="# Please pay $"+Double.valueOf(df.format(cost[0]))+" for the vms #";

		return output;
	}
	
	/**
	 * method purchaseVMs: processes client purchase of VMs for a job request
	 * @param jobRequest
	 * @param cost
	 * @return: output stream
	 */
	private String purchaseVMs(Job jobRequest, double [] cost) {

		String output="";
		for (int i=0; i<jobRequest.tasks.size(); i++) {
			TaskObj t=jobRequest.tasks.get(i);
			if(t.getVmType()>=0)
				output+=" VMtype :"+vmTypeList.get(t.getVmType()).getname()+" launched with vmid "+t.getVmId()+" for task:" +t.getTaskId() +"#";
		}
		
		DecimalFormat df = new DecimalFormat("0.00");									
		output+="# Please pay $"+Double.valueOf(df.format(cost[0]))+" for the vms #";

		return output;
	}
	
	/**
	 * method cancelVM: cancels vms in vmid list
	 * @param vmidList
	 * @return: output string
	 */
	private String cancelVMs(ArrayList<Integer> vmidList) {

		String output;
		System.out.println("vmidList size "+vmidList.size());
		while(vmidList.size()>0) {
			Integer vmid=vmidList.remove(0);
			Integer [] vmOpts=activeVmList.remove(vmid);
			System.out.println("Cancelling vm"+vmid+" on server"+serverList.get(vmOpts[0]).getid());			    				
			serverList.get(vmOpts[0]).terminateVM(vmid);
		}
		output="Cancelled Vms released back to the datacenter #";

		return output;
	}
	/**
	 * method connectToVM: connects client to vm resource vmid
	 * @param vmid
	 * @return: output string
	 */
	private String connectToVM(Integer vmid) {
		
		String output;
		if(activeVmList.containsKey(vmid)) {
			Integer [] vmOpts=activeVmList.get(vmid);
			if(vmOpts[2]==0) {
				Integer serverID=serverList.get(vmOpts[0]).getid();
				output=String.valueOf(serverID)+"?"+String.valueOf(vmid)+"?"+String.valueOf(vmOpts[1]);
				vmOpts[2]=1;
				activeVmList.put(vmid, vmOpts);
				System.out.println("vmid: "+vmid+" is now connected to user on server "+serverID);		    							    						
			} else
				output="vmid: "+vmid+" is currently connected to the user";
		} else
			output="vmid is invalid";

		return output;
	}

	/**
	 * method discoonectFromVM: disconnects client from vm resource vmid
	 * @param vmid
	 * @return: output string
	 */
	private String disconnectFromVM(Integer vmid) {
		
		String output;
		if(activeVmList.containsKey(vmid)) {
			Integer [] vmOpts=activeVmList.get(vmid);
			vmOpts[2]=0;
			activeVmList.put(vmid, vmOpts);
			System.out.println("vmid: "+vmid+" is now disconnected");		    			
			output="vmid: "+vmid+" is now disconnected";		    					
		} else
			output="vmid is invalid";
			
		return output;
	}

	/**
	 * method submitJob: submits job to a cloud
	 * @param jobRequest: job specification.  Altered by vm info assigned to each task
	 * @param lb: load balancer
	 * @param vmidList: list of vmids generated for the job
	 * @param cost: cost to run the job
	 * @return: output string
	 */
	private String submitJob(Job jobRequest, LoadBalancer lb, ArrayList<Integer> vmidList, double [] cost) {
		
		int id0=vmCtr;
		vmCtr+=jobRequest.tasks.size();
		
		//call load balancer for job placement
		ArrayList<Integer []> serverAllocation=lb.jobPlacement(jobRequest, serverList,id0);
		
		//estimate cost of assigned vms
		vmidList= new ArrayList<Integer> ();			    				
		while(serverAllocation.size()>0) {
			Integer [] a= serverAllocation.remove(0);
			cost[0]+=vmTypeList.get(a[1]).getprice();	
			System.out.println(a[0]+" "+a[1]+" "+a[2]);
			Integer [] vmOpts= {a[0], a[1], 0};
			activeVmList.put(a[2], vmOpts);
			vmidList.add(a[2]);			    					
		}
		String output="";
		for (int i=0; i<jobRequest.tasks.size(); i++) {
			TaskObj t=jobRequest.tasks.get(i);
			if(t.getVmType()>=0)
				output+=" VMtype :"+vmTypeList.get(t.getVmType()).getname()+" recommended for task:" +t.getTaskId() +"#";
		}
		DecimalFormat df = new DecimalFormat("0.00");							
		output+="# Total cost for vms: $"+Double.valueOf(df.format(cost[0]))+"#";
		
		return output;

	}

	/**
	 * method releaseVM: releases vm resource vmid 
	 * @param vmid
	 * @return output string
	 */
	private String releaseVM(Integer vmid) {
		
		String output;
		if(activeVmList.containsKey(vmid)) {
			Integer [] vmOpts=activeVmList.get(vmid);
			if(vmOpts[2]==0) {
				activeVmList.remove(vmid);
				System.out.println("Releasing vm"+vmid+" on server"+serverList.get(vmOpts[0]).getid());			    				
				serverList.get(vmOpts[0]).terminateVM(vmid);
				
				output="Vms released back to the datacenter #";
			} else
				output="vmid: "+vmid+" is currently connected to the user";
		} else
			output="vmid is invalid";
		
		return output;
		
	}
	
	// Executes a specific workflow given a definition specified by a workflow plan file
	// VM's are requested using type and specific ID numbers are assigned
	// Communication takes place between assigned VM's specifying task start/stop
	// VM are released following completion of workflow
	public void runWorkflowJob(String planFile) throws IOException {
		String s = "";
		String [] tokens;
		ArrayList<Server> serverList;			// Holds the list of available servers
		ArrayList<Integer> vmList = new ArrayList<Integer>(); // Holds the requested VM's
		ArrayList<Integer> idList = new ArrayList<Integer>(); // Holds VM id's
				
		// Read VM list
		//ArrayList<VirtualMachine> vmTypeList = buildVmList();

		// Query Data Center for available servers
		//serverList = DC.getServersList();

		BufferedReader in = new BufferedReader(new FileReader(planFile));
		int vmReady = 0;

		// Read first line containing 'Workflow Request'
		s = in.readLine();	// Read next line from the input text file

		if (s.toLowerCase().contains("workflow request")) {
			System.out.println("Workflow Plan File Opened");
		} else {
			System.out.println("Workflow Plan File Format Error");
			return;
		}

		// Add all of the edges to graph
		while (in.ready()) {
			s = in.readLine();	// Read next line from the input text file
			//System.out.println(s);
			tokens = s.split(",");
			//for (int i=0; i<tokens.length; i++) {
				//System.out.print(tokens[i] + ",");
			//}
			//System.out.print("\n");

			if (tokens[0].contains("vmGet")) {
				// Allocate a new vm
				if (vmReady == 0) {
					vmList.add(Integer.parseInt(tokens[1]));
					idList.add(Integer.parseInt(tokens[2]));
					System.out.println("vmGet: " + tokens[1] + ", " + tokens[2]);
				} else {
					System.out.println("VM's can no longer be requested. Aborting ...");
					return;
				}
			} else if (tokens[0].contains("vmReturn")) {
				// Deallocate vm's
				if (vmReady == 1) {
					vmReady = 2;
				}
				System.out.println("vmReturn: " + tokens[1]);
			} else if (tokens[0].contains("vm")) {
				// Command VM's to do something
				if (vmReady == 0) {
					vmReady = 1;
				} else if (vmReady == 2) {
					System.out.println("VM's can no longer be accessed. Aborting ...");
					return;
				}
				System.out.println("vm: " + tokens[1] + ", " + tokens[2]);
			} else if (tokens[0].contains("pause")) {
				// Pause execution to simulate network delays, should we just add this time to
				// the execution time of each vm?
				System.out.println("pause: " + tokens[1]);
			} else {
				System.out.println(tokens[0] + ": is an invalid command. Skipping...");
			}
		}
	}
}
